package com.atguigu.flink.chapter03_exec1;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.flink.chapter02_DataStreamAPI.source.MySource;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2022/10/24
 *
 *  {"behavior":"download","channel":"HUAWEI","timestamp":1666572813450,"userId":1402}
 */
public class Demo3_Channel
{
    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        //自定义数据源
        MySource mySource = new MySource();

        env
            .addSource(mySource)
            .map(new MapFunction<String, Tuple2<String,Integer>>()
            {
                @Override
                public Tuple2<String, Integer> map(String value) throws Exception {

                    //转换为Map集合  JSONObject extends JSON implements Map<String, Object>
                    JSONObject jsonObject = JSON.parseObject(value);
                    String key = jsonObject.getString("channel").concat("-").concat(jsonObject.getString("behavior"));

                    return Tuple2.of(key,1);
                }
            })
            .keyBy(new KeySelector<Tuple2<String, Integer>, String>()
            {
                @Override
                public String getKey(Tuple2<String, Integer> value) throws Exception {
                    return value.f0;
                }
            })
            .sum(1)
            .print();


        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }
    }
}
